package com.yanxu;

import com.yanxu.domain.Event;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.util.Collector;

import java.util.Calendar;
import java.util.Random;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025-07-13-2025
 * @Description: api 学习
 *  reduce 归于聚合操作
 */
public class Api10_reduce {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //源 -算子
        // 使用 addSource 自定义数据源
        DataStreamSource<Event> streamSource = env.addSource(new ParallelSourceFunction<Event>() {
            boolean running = true;

            @Override
            public void run(SourceContext<Event> ctx) throws Exception {
                // 创建Random实例，用于生成随机数
                Random random = new Random();
                // 定义用户数组和访问URL数组
                String[] users = {"Mary", "Alice", "Bob", "Cary"};
                String[] urls = {"./home", "./cart", "./fav", "./prod?id=1", "./prod?id=2"};

                // 当running为true时，持续生成事件
                while (running) {
                    // 随机选择一个用户和URL，结合当前时间戳，创建一个新的Event对象并收集
                    ctx.collect(new Event(
                            users[random.nextInt(users.length)],
                            urls[random.nextInt(urls.length)],
                            Calendar.getInstance().getTimeInMillis()
                    ));
                    // 为了便于观察和处理，每生成一个事件后暂停1秒
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        });

        streamSource.print("add source >>>>");

        // 使用 flatmap 进行转换
        SingleOutputStreamOperator<Tuple3<String, Integer, Integer>> flatMapOperator = streamSource.flatMap(new FlatMapFunction<Event, Tuple3<String, Integer, Integer>>() {
            @Override
            public void flatMap(Event event, Collector<Tuple3<String, Integer, Integer>> collector) throws Exception {
                String name = event.getName();
                collector.collect(new Tuple3<>(name, 1, new Random().nextInt(10000)));
            }
        });

        //使用keyby 进行分区
        KeyedStream<Tuple3<String, Integer, Integer>, String> KeyedStream =
                flatMapOperator.keyBy(tuple -> tuple.f0);

        //自定义 reduceFunction 自己实现 sum 的效果
        SingleOutputStreamOperator<Tuple3<String, Integer, Integer>> reduceOperator = KeyedStream.reduce(new ReduceFunction<Tuple3<String, Integer, Integer>>() {
            /*
             * value1 : 上一次reduce 返回的结果
             * value2 : 本次的输入的 tuple
             * */
            @Override
            public Tuple3<String, Integer, Integer> reduce(Tuple3<String, Integer, Integer> value1,
                                                           Tuple3<String, Integer, Integer> value2) throws Exception {
                return new Tuple3<>(value1.f0, value1.f1 + value2.f1, null);
            }
        });
        reduceOperator.print("reduce function >>>");


        //自定义 reduceFunction 自己实现 max 的效果
        SingleOutputStreamOperator<Tuple3<String, Integer, Integer>> reduceOperator2 = KeyedStream.reduce(new ReduceFunction<Tuple3<String, Integer, Integer>>() {
            @Override
            public Tuple3<String, Integer, Integer> reduce(Tuple3<String, Integer, Integer> value1,
                                                           Tuple3<String, Integer, Integer> value2) throws Exception {
                Integer num = value1.f2 > value2.f2 ? value1.f2 : value2.f2;
                return new Tuple3<>(value1.f0, null, num);
            }
        });
        reduceOperator2.print("reduce2 function >>>");

        //对环境进行执行。
        env.execute();
    }
}